package com.wjw.rabbit.direct;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 路由模式：direct,有路由Key
 * 消费者(接收消息)
 */
public class Consumer {

    public static Runnable runnable =  new Runnable() {
        public void run() {
            //1.创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //2.设置连接属性
            factory.setHost("127.0.0.1");//RabbitMQ服务的IP
            factory.setPort(5672);//端口
            factory.setUsername("admin");//RabbitMQ的用户
            factory.setPassword("admin");//RabbitMQ的密码
            factory.setVirtualHost("/");//虚拟访问节点
            //获取队列名称
            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                //3.从工厂中获取连接
                connection = factory.newConnection("消费者");
                //4.从连接中获取通道
                channel = connection.createChannel();
                /**
                 * 声明队列queue存储消息
                 * 如果队列不存在,则会创建
                 * 参数1:queue队列名称
                 * 参数2：durable队列是否持久化
                 * 参数3：exclusive是否排他，即是否私有，如果为true，会对当前队列加锁，其他通道不能访问，并且连接自动关闭
                 * 参数4：autoDelete:是否自动删除，当最后一个消费者断开连接之后，是否自动删除消息
                 * 参数5：arguments，可以设置队列附件参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等
                 */
                //这里如果queue已经被创建过一次了，可以不需要定义
                //channel.queueDeclare("queue1",false,false,false,null);
                Channel finalChannel = channel;
                channel.basicConsume(queueName, true, new DeliverCallback() {
                    //@Override
                    public void handle(String s, Delivery delivery) throws IOException {
                        System.out.println(delivery.getEnvelope().getDeliveryTag());
                        System.out.println(queueName+":收到的消息是："+new String(delivery.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    public void handle(String s) throws IOException {

                    }
                });
                System.out.println(queueName+":开始接收消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                //7.关闭通道
                if(channel != null && channel.isOpen()){
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                //8.关闭连接
                if(connection != null && connection.isOpen()){
                    try {
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }//run
    };//runnable

    public static void main(String[] args) {
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
    }
}
